package com.afdor.rws.event.driven.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.driven.EventPersistenceHandler;
import com.afdor.rws.event.driven.EventPersistenceManager;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.config.EventPersistenceConfig;
import com.afdor.rws.event.driven.entities.Event;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 事件持久化管理器
 *
 * @author 悭梵
 * @date Created in 2018-08-20 14:25
 */
@Slf4j
@Extension(value = "default,group", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventPersistenceManager.class)
public class DefaultEventPersistenceManager implements EventPersistenceManager, Runnable {
    /**
     * 事件持久化管理器一次持久化事件数量
     */
    @Getter
    private int batchSize = 1;
    /**
     * 配置新仙尼
     */
    @Getter
    private EventPersistenceConfig config;
    @Getter
    private List<EventPersistenceHandler> eventPersistenceHandlers;
    @Getter
    private ConcurrentLinkedQueue<Event> eventQueue = new ConcurrentLinkedQueue<>();

    private AtomicInteger continuousException = new AtomicInteger(0);


    @Override
    public <T extends EventPersistenceManager, C extends EventPersistenceConfig> T setConfig(C config) {
        this.config = config;
        return (T) this;
    }

    @Override
    public synchronized <T extends EventPersistenceManager> T addEventPersistenceHandler(@NonNull EventPersistenceHandler persistenceHandler) {
        if (getEventPersistenceHandlers() == null) {
            this.eventPersistenceHandlers = new ArrayList<>(10);
        }
        getEventPersistenceHandlers().add(persistenceHandler);
        return (T) this;
    }

    /**
     * 添加事件
     *
     * @param events
     */
    @Override
    public <T extends EventPersistenceManager> T addEvent(Event... events) {
        // 事件持久化处理器为空，不进行事件持久化
        if (Objects.isNull(getEventPersistenceHandlers()) || getEventPersistenceHandlers().isEmpty()) {
            return (T) this;
        }

        for (int i = 0, length = events.length; i < length; i++) {
            // 添加事件
            getEventQueue().offer(events[i]);
        }
        return (T) this;
    }


    @Override
    public void run() {
        if (!Thread.currentThread().isInterrupted()) {
            /* 线程未终止，就执行事件消费。若持久化事件返回为true，这继续循环处理事件； 若处理事件返回为false，则退出执行，有计划任务重新启动执行*/
            while (consume()) {
                if (Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    @Override
    public <T extends EventPersistenceManager> T init() {

        batchSize = getConfig().getBatchSize();
        if (batchSize <= 0) {
            batchSize = 1;
        }
        int period = getConfig().getPeriod();
        if (period <= 0) {
            period = 1000;
        }
        ThreadUtils.getScheduledThreadPool().scheduleWithFixedDelay(this, 0, period, TimeUnit.MILLISECONDS);
        return (T) this;
    }

    @Override
    public void destroy() {
        ThreadUtils.getScheduledThreadPool().remove(this);
    }

    /**
     * 执行具体事件持久化操作
     *
     * @return
     */
    protected boolean consume() {
        // 允许事件持久化处理器为空，为空即表示事件的持久化不由系统复杂
        List<EventPersistenceHandler> eventPersistenceHandlerList = getEventPersistenceHandlers();
        // 存在持久化处理器时，执行持久化处理逻辑
        if (!(eventPersistenceHandlerList == null || eventPersistenceHandlerList.isEmpty())) {
            if (getBatchSize() <= 1) {
                Event event = getEventQueue().poll();
                if (event != null) {
                    eventPersistenceHandlerList.forEach(eventPersistenceHandler -> {
                        try {
                            eventPersistenceHandler.persistence(event);
                            continuousException.getAndSet(0);
                        } catch (Exception e) {
                            addEvent(event);
                            if (continuousException.incrementAndGet() > 3) {
                                // TODO 连续持久化异常警告，及处理策略
                            } else if (continuousException.incrementAndGet() > 5) {
                                // TODO 连续持久化异常警告，及处理策略
                            }
                            // 若这里持续报异常，可能会出现事件队列满溢出的BUG或频繁GC，导致OutOfMemoryError。
                            log.error("持久化事件异常，已将事件重新放入队列，稍后继续执行持久化", e);
                        }
                    });
                }
            } else {
                int batchSize = getBatchSize();
                /* 获取本次需要持久化的事件列表 */
                final List<Event> batchEvent = new ArrayList<>(batchSize);
                while (Boolean.TRUE) {
                    Event event = getEventQueue().poll();
                    if (event != null) {
                        batchEvent.add(event);
                        if (batchEvent.size() >= batchSize) {
                            break;
                        }
                    } else {
                        break;
                    }
                }

                eventPersistenceHandlerList.forEach(eventPersistenceHandler -> {
                    try {
                        eventPersistenceHandler.persistence(batchEvent);
                        continuousException.getAndSet(0);
                    } catch (Exception e) {
                        addEvent((Event[]) batchEvent.toArray());
                        if (continuousException.incrementAndGet() > 3) {
                            // TODO 连续持久化异常警告，及处理策略
                        } else if (continuousException.incrementAndGet() > 5) {
                            // TODO 连续持久化异常警告，及处理策略
                        }
                        // 若这里持续报异常，可能会出现事件队列满溢出的BUG或频繁GC，导致OutOfMemoryError。
                        log.error("持久化事件异常，已将事件重新放入队列，稍后继续执行持久化", e);
                    }
                });
            }
        } else {
            // 不存在持久化处理器时，直接清空队列
            getEventQueue().clear();
        }
        return !getEventQueue().isEmpty();
    }
}
